package com.ironman.imes.device.data.source.verticles;

import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import com.ironman.imes.device.data.source.BaseSource;
import com.ironman.imes.device.event.model.DeviceSourceData;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttClientOptions;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.noear.snack.ONode;

import java.util.List;


@Slf4j
@RequiredArgsConstructor
public class MqttClientVerticle extends AbstractVerticle implements BaseSource {

    private final String name;

    private final String host;

    private final int port;

    private final String clientIdPrefix;

    private final String username;

    private final String password;

    private final boolean cleanSession;

    private final int keepAliveInterval;

    private final List<ONode> topics;

    private MqttClient mqttClient;

    @Override
    public void start(Promise<Void> startPromise) {
        this.mqttClient = io.vertx.mqtt.MqttClient.create(vertx, new MqttClientOptions()
                .setClientId(clientIdPrefix + RandomUtil.randomNumbers(6))
                .setUsername(username).setPassword(password)
                .setCleanSession(cleanSession)
                .setKeepAliveInterval(keepAliveInterval));
        mqttClient.connect(port, host, connectHandler -> {
            if (connectHandler.succeeded()) {
                log.info("【{}】MQTT服务端连接成功", name);
                for (ONode topicNode : topics) {
                    String topicName = topicNode.get("topic").getString();
                    ONode meta = topicNode.get("meta");
                    mqttClient.publishHandler(message -> {
                        String topic = message.topicName();
                        String payload = message.payload().toString();
                        log.info("【{}】MQTT收到消息：{} {}", name, topic, payload);

                        String deviceType = null;
                        String deviceTypeKey = meta.select("$.deviceType.index").getString();
                        if (StrUtil.isNotBlank(deviceTypeKey)) {
                            deviceType = topic.split("/")[Integer.parseInt(deviceTypeKey)];
                        }
                        deviceTypeKey = meta.select("$.deviceType.body").getString();
                        if (StrUtil.isNotBlank(deviceTypeKey)) {
                            deviceType = ONode.loadStr(payload).select(deviceTypeKey).getString();
                        }

                        String deviceCode = null;
                        String deviceCodeKey = meta.select("$.deviceCode.index").getString();
                        if (StrUtil.isNotBlank(deviceCodeKey)) {
                            deviceCode = topic.split("/")[Integer.parseInt(deviceCodeKey)];
                        }
                        deviceCodeKey = meta.select("$.deviceCode.body").getString();
                        if (StrUtil.isNotBlank(deviceCodeKey)) {
                            deviceCode = ONode.loadStr(payload).select(deviceCodeKey).getString();
                        }
                        launchData(new DeviceSourceData(topicName, deviceType, deviceCode, payload));
                    }).subscribe(topicName, topicNode.get("qos").getInt());
                }
            } else {
                log.error("MQTT客户端连接失败 {}", name, connectHandler.cause());
            }
        });
    }

    @Override
    public void stop(Promise<Void> stopPromise) {
        if (mqttClient != null && mqttClient.isConnected()) {
            mqttClient.disconnect();
        }
    }
}